package com.bml.architect.kafka;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("/kafka")
public class KafkaControler {

    private String topic = "my-topic";

    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;

    /**
     * 同步发送,发送就不管了
     * @param key
     * @param msg
     * @return
     */
    @RequestMapping("/send")
    public String send(@RequestParam(name = "key")String key,@RequestParam(name = "msg")String msg){
        ListenableFuture<SendResult<String, String>> send = kafkaTemplate.send(topic, 0, key, msg);
        return "send is success!";
    }

    @RequestMapping("/async")
    public String async(@RequestParam(name = "key")String key,@RequestParam(name = "msg")String msg){
        kafkaTemplate.send(topic, 0, key, msg).addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable ex) {
                System.out.println("消息发送失败");

            }

            @Override
            public void onSuccess(SendResult<String, String> result) {
                System.out.println("消息发送成功:" + result.getRecordMetadata().topic());
            }
        });
        return "send async is success!";
    }
}
